万得EDB数据拉取写入holo
import requests
import json
import psycopg2
import math
import pymysql

from requests.exceptions import RequestException, Timeout

def WindEebDataToDb():
    edbCodes=getParam(edbCodes)
    beginDate=getParam(beginDate)
    endDate=getParam(endDate)
    # edbCodes='M0048666,M0048665,M0048660,M0048657,M0048659,M0048661,M0048663,M0048662,M0048664,M5567876,M0096212,M0000553,M0096211'
    # beginDate='2000-01-01'
    # endDate='2024-11-18'
    edbCodesList = edbCodes.split(',')
    # 初始化一个空列表来存储分组
    # 初始化一个空列表来存储分组后的字符串
    grouped_list = []
    print("wind每日刷新指标列表:"+edbCodes)
    # 遍历列表，每次取两个元素进行分组
    for i in range(0, len(edbCodesList), 50):
        # 取出当前索引和下一个索引的元素，如果下一个索引超出范围，则只取当前索引的元素
        pair = edbCodesList[i:i+50]
        # 将当前组的元素用逗号连接，并添加到结果列表中
        grouped_list.append(','.join(pair))
    # 打印结果列表
    print("wind指标分组:"+str(grouped_list))
    records=[]
    metas=[]
    for codes in grouped_list:
        refreshMetrics=[]
        try:
            url = 'http://10.74.194.151:8832/windapi?apiName=edb&edbCodes='+codes+'&originalResult=true&param=field_info=True'
            # 发送POST请求
            response = requests.post(url, data={})
            json_data = json.loads(response.text)
            if json_data['code']==0:
                for j in range(len(json_data["data"][0])):
                    metrics_update=json_data["data"][6][j]
                    # 采集更新时间大于开始时间
                    if metrics_update>beginDate:
                        metas.append({
                            "edb_metrics_code": "WD-"+json_data["data"][0][j],
                            "third_code": json_data["data"][0][j],
                            "third_type":"WD",
                            "metrics_name": json_data["data"][1][j],
                            "metrics_frequence": json_data["data"][2][j],
                            "metrics_unit": json_data["data"][3][j],
                            "metrics_startdate": json_data["data"][4][j],
                            "metrics_enddate": json_data["data"][5][j],
                            "metrics_update": json_data["data"][6][j],
                            "metrics_data_source": json_data["data"][7][j],
                            "edb_catagory": json_data["data"][8][j],
                            "metrics_remark": json_data["data"][9][j],
                            "metrics_nation": json_data["data"][10][j]
                        })
                        refreshMetrics.append(json_data["data"][0][j])
                    else:
                        print("不用更新的wind数据指标:"+json_data["data"][0][j]+",最新更新时间:"+metrics_update)
        except Timeout:
            # 捕获超时异常
            print("WDAPI元数据请求请求超时")
        except RequestException as e:
            # 捕获其他请求相关的异常
            print(f"WDAPI元数据请求请求异常: {e}")
        else:
            # 如果没有异常发生，处理响应
            print("WDAPI元数据请求成功")
        if len(refreshMetrics)>0:
            try:
                dataCodes = ','.join(refreshMetrics)
                url = 'http://10.74.194.151:8832/windapi?apiName=edb&edbCodes='+dataCodes+'&beginDate='+beginDate+'&endDate='+endDate+'&originalResult=true'
                # 发送POST请求
                response = requests.post(url, data={})
                data = json.loads(response.text)

                if data['code']==0:
                    datas=data['data']
                    times=data['times']
                    codes=data['codes']
                    for index, value in enumerate(codes):
                        dataList=datas[index]
                        for dindex, dvalue in enumerate(dataList):
                            records.append({
                                "keys": 'WIND-' +value + times[dindex],  # 主键
                                "edb_metrics_code": 'WD-' + value,  # 主键
                                "metrics_value": dvalue,  # 值可以是任何数据类型
                                "data_time": times[dindex],  # 确保使用正确的键名和字符串值
                                "third_code": value,
                                "third_type": "WD"
                            })
            except Timeout:
                # 捕获超时异常
                print("WDAPI请求超时")
            except RequestException as e:
                # 捕获其他请求相关的异常
                print(f"WDAPI请求异常: {e}")
            else:
                # 如果没有异常发生，处理响应
                print("WDAPI请求成功")
    values = [d['third_code'] for d in metas]
    result = ','.join(values)
    print(beginDate+"之后有更新的wind数据指标:"+result)
    print("wind元数据查询完成,查询条数:"+str(len(metas)))
    print("wind指标值数据数据查询完成,查询条数:"+str(len(records)))
    conn = psycopg2.connect(host="hgprecn-cn-v641lnkxm003-cn-shanghai.hologres.aliyuncs.com",
                            port=80,
                            dbname="odpstest",
                            user="LTAI5tQo9VtJ414iSEZrE8Vn",
                            password="BPBKqDiQ7JKPl6o3QDIFg6kh71nakS",
                            application_name="third_edb_data")
    # 使用executemany执行批量插入
    try:
        # 导入数据
        cur = conn.cursor()
        print("开始获取连接")
        # 初始化一个列表来存储批量插入的数据
        # 插入元数据
        insertlen=0;
        batch = []
        # 初始化一个计数器
        counter = 0;
        for data in metas:
            # 为每条记录创建一个元组，并添加到批量列表中
            batch.append((
                data['edb_metrics_code'],
                data['edb_catagory'],
                data['third_code'],
                data['third_type'],
                data['metrics_name'],
                data['metrics_unit'],
                data['metrics_frequence'],
                data['metrics_remark'],
                data['metrics_data_source'],
                data['metrics_startdate'],
                data['metrics_enddate'],
                data['metrics_update'],
                data['metrics_nation']
            ))
            # 每收集500条记录，执行一次插入操作
            if counter >= 500:
                # 构造批量插入的SQL语句
                values = ", ".join(["(%s, %s, %s, %s, %s,%s, %s,%s, %s, %s, %s, %s,%s,NUll)" for _ in range(len(batch))])
                sql = (f"INSERT INTO public.index_center_third_edb_metrics_main(edb_metrics_code, edb_catagory, third_code, third_type,metrics_name,"
                       f"metrics_unit,metrics_frequence,metrics_remark,metrics_data_source,metrics_startdate,metrics_enddate,metrics_update,metrics_nation,available_date) "
                       f"VALUES {values} ON CONFLICT (edb_metrics_code) DO UPDATE SET metrics_startdate = EXCLUDED.metrics_startdate,metrics_update = EXCLUDED.metrics_update,metrics_enddate = EXCLUDED.metrics_enddate,"
                       f"metrics_unit = EXCLUDED.metrics_unit,metrics_frequence = EXCLUDED.metrics_frequence,"
                       f"metrics_remark = EXCLUDED.metrics_remark,metrics_data_source = EXCLUDED.metrics_data_source, modify_time = now()")

                # 执行批量插入
                cur.execute(sql, [item for sublist in batch for item in sublist])
                # 提交事务
                conn.commit()
                insertlen+= 500
                # 重置计数器和批量列表
                counter = 0
                batch = []
                # 更新计数器
            counter += 1

        # 如果最后一批记录不足500条，也进行插入和提交
        if batch:
            values = ", ".join(["(%s, %s, %s, %s, %s,%s, %s,%s, %s, %s, %s, %s,%s,NUll)" for _ in range(len(batch))])
            sql = (f"INSERT INTO public.index_center_third_edb_metrics_main(edb_metrics_code, edb_catagory, third_code, third_type,metrics_name,"
                   f"metrics_unit,metrics_frequence,metrics_remark,metrics_data_source,metrics_startdate,metrics_enddate,metrics_update,metrics_nation,available_date) "
                   f"VALUES {values} ON CONFLICT (edb_metrics_code) DO UPDATE SET metrics_startdate = EXCLUDED.metrics_startdate,metrics_update = EXCLUDED.metrics_update,metrics_enddate = EXCLUDED.metrics_enddate,"
                   f"metrics_unit = EXCLUDED.metrics_unit,metrics_frequence = EXCLUDED.metrics_frequence,"
                   f"metrics_remark = EXCLUDED.metrics_remark,metrics_data_source = EXCLUDED.metrics_data_source, modify_time = now()")
            # 执行批量插入
            cur.execute(sql, [item for sublist in batch for item in sublist])
            # 提交事务
            conn.commit()
            insertlen+= len(batch)
        print("元数据数据写入完成,写入条数:"+str(insertlen))
        # 插入数据
        insertlen=0;
        batch = []
        # 初始化一个计数器
        counter = 0
        # 准备批量插入的数据
        for data in records:
            # 为每条记录创建一个元组，并添加到批量列表中
            value = float(data['metrics_value'])
            if not math.isnan(value):
                batch.append((
                    data['keys'],
                    data['edb_metrics_code'],
                    data['metrics_value'],
                    data['data_time'],
                    data['third_code'],
                    data['third_type']
                ))
            # 每收集500条记录，执行一次插入操作
            if counter >= 500:
                # 构造批量插入的SQL语句
                values = ", ".join(["(%s, %s, %s, %s, %s,%s)" for _ in range(len(batch))])
                sql = f"INSERT INTO public.index_center_third_edb_metrics_data(keys, edb_metrics_code, metrics_value, data_time,third_code,third_type) VALUES {values} ON CONFLICT (keys) DO UPDATE SET metrics_value = EXCLUDED.metrics_value, modify_time = now()"

                # 执行批量插入
                cur.execute(sql, [item for sublist in batch for item in sublist])
                # 提交事务
                conn.commit()
                insertlen+= 500
                # 重置计数器和批量列表
                counter = 0
                batch = []
                # 更新计数器
            counter += 1

        # 如果最后一批记录不足500条，也进行插入和提交
        if batch:
            values = ", ".join(["(%s, %s, %s, %s, %s,%s)" for _ in range(len(batch))])
            sql = f"INSERT INTO public.index_center_third_edb_metrics_data(keys, edb_metrics_code, metrics_value, data_time,third_code,third_type) VALUES {values} ON CONFLICT (keys) DO UPDATE SET metrics_value = EXCLUDED.metrics_value, modify_time = now()"
            # 执行批量插入
            cur.execute(sql, [item for sublist in batch for item in sublist])
            # 提交事务
            conn.commit()
            insertlen+= len(batch)

    except Exception as e:
        print("发生错误：", e)
        # 发生错误时回滚
        conn.rollback()
    finally:
        # 关闭游标和连接
        cur.close()
        conn.close()
    print("指标值数据数据写入完成")
    update_queries = []
    for data in metas:
        index_code = data['edb_metrics_code']
        end_date = data['metrics_enddate']
        update_query = f"""
        UPDATE data_browser.edb_index_center
        SET index_update_time = '{end_date}'
        WHERE index_code = '{index_code}'
        """
        update_queries.append(update_query)
    # 执行更新操作
    update_db('rds50g3807a68zwc9soo969.mysql.rds.aliyuncs.com', 'data_browser', 3306, 'D6a_T3_Brs', update_queries)
    print("Update completed successfully.")

def update_db(host, user, port, password, update_queries):
    conn = pymysql.connect(host=host, user=user, port=port, password=password, charset='utf8')
    try:
        c = conn.cursor()
        for query in update_queries:
            c.execute(query)
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()
if __name__ == '__main__':
    WindEebDataToDb()

